
#include"threadpool.h"
#include<pthread.h>
#include<stdio.h>
#include<malloc.h>
#include<unistd.h>
#include<stdlib.h>

const int NUMBER = 2;
//任务结构体
typedef struct Task{
    void (*func)(void* arg);
    void* arg;
}Task;
//线程池结构体
typedef struct ThreadPool{
    //任务队列
    Task* taskQ;
    int queueCapacity;//容量
    int queueSize;//当前任务个数
    int queueFront;	//对头->取数据
    int queueRear;	//队尾->放数据

    pthread_t managerID;	//管理者线程ID
    pthread_t *threadIDs;	//工作线程ID
    int minNum;	//最小线程数
    int maxNum;	//最大线程数
    int busyNum;	//忙的线程数==工作的线程
    int liveNum;	//存活的线程数
    int exitNum;	//要杀死的线程个数
    pthread_mutex_t mutexPool;	//互斥锁，锁整个线程池
    pthread_mutex_t mutexBusy;	//锁busyNum变量，对于灵活修改的变量，需要考虑单独上锁，操作效率会比较高
    //利用条件变量负责阻塞生产者和消费者
    pthread_cond_t notFull;	//任务队列是不是满了
    pthread_cond_t notEmpty;	//任务队列是不是空了

    int shutdown;	//是不是要销毁线程池，销毁1，不销毁为0
}ThreadPool;
//创建线程池并初始化
ThreadPool* threadPoolCreate(int min, int max, int queueSize) {
    ThreadPool* pool = (ThreadPool *)malloc(sizeof(ThreadPool));
    do {
        if (pool == NULL) {
            printf("malloc threadpool fail\n");
            break;
        }
        pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);
        if (pool->threadIDs == NULL) {
            printf("malloc threadIDs fail\n");
            break;
        }
        memset(pool->threadIDs, 0, sizeof(pthread_t) * max);
        pool->minNum = min;
        pool->maxNum = max;
        pool->busyNum = 0;
        pool->liveNum = min;	//和最小个数相等
        pool->exitNum = 0;
        if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
            pthread_cond_init(&pool->notEmpty, NULL) != 0 ||
            pthread_cond_init(&pool->notFull, NULL) != 0) {
            printf("mutex or condition init fail\n");
            break;
        }
        //任务队列
        pool->taskQ = (Task*)malloc(sizeof(Task) * queueSize);
        pool->queueCapacity = queueSize;
        pool->queueSize = 0;
        pool->queueFront = 0;
        pool->queueRear = 0;
        pool->shutdown = 0;

        //创建线程
        pthread_create(&pool->managerID, NULL, manager, pool);
        for (int i = 0; i < min; i++) {
            pthread_create(&pool->threadIDs[i], NULL, worker, pool);
        }
        return pool;
    } while (0);
    //释放资源
    if (pool&&pool->threadIDs) {
        free(pool->threadIDs);
    }
    if (pool&&pool->taskQ) {
        free(pool->taskQ);
    }
    if (pool) {
        free(pool);
    }
    return NULL;
}

int threadPoolDestroy(ThreadPool* pool){
    if (pool == NULL) {//指向的不是一块有效的地址
        return -1;
    }
    //关闭线程池
    pool->shutdown = 1;
    //阻塞回收管理者线程
    pthread_join(pool->managerID, NULL);
    //唤醒阻塞的消费者线程
    for (int i = 0; i < pool->liveNum; i++) {
        pthread_cond_signal(&pool->notEmpty);
    }
    //释放堆内存
    if (pool->taskQ) {
        free(pool->taskQ);
    }
    if (pool->threadIDs) {
        free(pool->threadIDs);
    }
    pthread_mutex_destroy(&pool->mutexPool);
    pthread_mutex_destroy(&pool->mutexBusy);
    pthread_cond_destroy(&pool->notEmpty);
    pthread_cond_destroy(&pool->notFull);
    free(pool);
    pool = NULL;
    return 0;
}

void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg){
    pthread_mutex_lock(&pool->mutexPool);
    while (pool->queueSize == pool->queueCapacity && pool->shutdown == 0) {
        //阻塞生产者线程
        pthread_cond_wait(&pool->notFull, &pool->mutexPool);
    }
    if (pool->shutdown) {
        pthread_mutex_unlock(&pool->mutexPool);
        return;
    }
    //添加任务
    pool->taskQ[pool->queueRear].func = func;
    pool->taskQ[pool->queueRear].arg = arg;
    pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity;//循环队列
    pool->queueSize++;

    pthread_cond_signal(&pool->notEmpty);
    pthread_mutex_unlock(&pool->mutexPool);
}

int threadPoolBusyNum(ThreadPool* pool){
    pthread_mutex_lock(&pool->mutexBusy);
    int busyNum = pool->busyNum;
    pthread_mutex_unlock(&pool->mutexBusy);
    return busyNum;
}

int threadPoolAliveNum(ThreadPool* pool){
    pthread_mutex_lock(&pool->mutexPool);
    int liveNum = pool->liveNum;
    pthread_mutex_unlock(&pool->mutexPool);
    return liveNum;
}

void* worker(void* arg)
{
    ThreadPool* pool = (ThreadPool*)arg;
    while (1){
        pthread_mutex_lock(&pool->mutexPool);
        //当前任务队列是否为空
        //必须要写while循环的原因：
        //假设有5个线程被唤醒，其中的某一个线程判断队列不为空，消费完了之后，
        //任务队列又为空，其他线程需要继续堵塞在这
        while (pool->queueSize==0&&!pool->shutdown) {
            //阻塞工作线程
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);

            //判断是不是要销毁线程
            if (pool->exitNum > 0) {
                pool->exitNum--;
                if (pool->liveNum > pool->minNum) {
                    pool->liveNum--;
                    pthread_mutex_unlock(&pool->mutexPool);
                    threadExit(pool);
                }
            }
        }
        //判断线程池是否被关闭了
        if (pool->shutdown) {
            pthread_mutex_unlock(&pool->mutexPool);
            //pthread_exit(NULL);
            threadExit(pool);
        }
        //从任务队列中取出一个任务
        Task task;
        task.func = pool->taskQ[pool->queueFront].func;
        task.arg = pool->taskQ[pool->queueFront].arg;
        //移动头节点（循环队列）
        pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity;//环形队列，取余实现
        pool->queueSize--;
        //解锁
        pthread_cond_signal(&pool->notFull);//生产者对应的条件变量
        pthread_mutex_unlock(&pool->mutexPool);

        printf("thread %ld start working\n",pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum++;
        pthread_mutex_unlock(&pool->mutexBusy);
        //执行
        task.func(task.arg);
        free(task.arg);
        task.arg = NULL;
        printf("thread %ld end working\n", pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexBusy);
    }
    return NULL;
}

void* manager(void* arg)
{
    ThreadPool* pool = (ThreadPool*)arg;
    while (!pool->shutdown) {
        //每隔3s检测一次
        sleep(3);
        //取出线程池中任务的数量和当前线程的数量
        pthread_mutex_lock(&pool->mutexPool);
        int queueSize = pool->queueSize;
        int liveNum = pool->liveNum;
        pthread_mutex_unlock(&pool->mutexPool);

        //取出忙的线程数量
        pthread_mutex_lock(&pool->mutexBusy);
        int busyNum = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexBusy);

        //添加线程
        //任务的个数大于存活的线程个数 && 存活的线程数小于最大线程数
        if (queueSize > liveNum && liveNum < pool->maxNum) {
            pthread_mutex_lock(&pool->mutexPool);
            int counter = 0;
            for (int i = 0; i < pool->maxNum && counter < NUMBER && pool->liveNum < pool->maxNum;i++) {
                if (pool->threadIDs[i]==0) {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);
                    counter++;
                    pool->liveNum++;
                }
            }
            pthread_mutex_unlock(&pool->mutexPool);
        }

        //销毁线程
        //忙的线程数乘以2小于存活的线程数 && 存活的线程数大于最小线程数
        if (busyNum * 2 < liveNum && liveNum > pool->minNum) {
            pthread_mutex_lock(&pool->mutexPool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexPool);
            //让工作线程自杀
            for (int i = 0; i < NUMBER; i++) {
                pthread_cond_signal(&pool->notEmpty);
                //把堵塞在条件变量的线程唤醒，向下执行，并让它们自杀
                /*
                //判断是不是要销毁线程
                if (pool->exitNum > 0) {
                    pool->exitNum--;
                    if (pool->liveNum > pool->minNum) {
                        pool->liveNum--;
                        pthread_mutex_unlock(&pool->mutexPool);
                        threadExit(pool);
                    }
                }
                */
            }
        }
    }
    return NULL;
}

void threadExit(ThreadPool* pool)
{
    //获取当前线程的id
    pthread_t tid = pthread_self();
    for (int i = 0; i < pool->maxNum; i++) {
        if (pool->threadIDs[i] = tid) {
            pool->threadIDs[i] = 0;
            printf("threadExit() called, %ld exiting\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

